package com.easylinkin.linkappapi.jzsmokewarning.service.impl;

import cn.hutool.core.lang.Assert;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.easylinkin.bases.redis.util.RedisUtil;
import com.easylinkin.linkappapi.common.utils.DateUtil;
import com.easylinkin.linkappapi.device.entity.Device;
import com.easylinkin.linkappapi.device.service.DeviceService;
import com.easylinkin.linkappapi.deviceunit.entity.DeviceUnit;
import com.easylinkin.linkappapi.deviceunit.service.DeviceUnitService;
import com.easylinkin.linkappapi.elasticsearch.entity.ESconfig;
import com.easylinkin.linkappapi.elasticsearch.util.EsCalendar;
import com.easylinkin.linkappapi.jzsmokewarning.entity.SmokeWarning;
import com.easylinkin.linkappapi.jzsmokewarning.mapper.TimingWarningMapper;
import com.easylinkin.linkappapi.jzsmokewarning.service.SmokeWarningService;
import com.easylinkin.linkappapi.jzsmokewarning.service.TimingWarning;
import com.easylinkin.linkappapi.openapi.dto.DatapushDTO;
import com.easylinkin.linkappapi.openapi.service.DataPushService;
import com.easylinkin.linkappapi.tenant.entity.LinkappTenant;
import com.easylinkin.linkappapi.tenant.mapper.LinkappTenantMapper;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.lucene.util.NamedThreadFactory;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.ParsedAvg;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.Test;
import org.quartz.Job;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 荆州油烟的定时预警
 * </p>
 *
 * @author TongJie
 * @since 2021-12-17
 */
@Service
public class TimingWarningImpl extends ServiceImpl<TimingWarningMapper, TimingWarning> implements TimingWarning, Job {

    /**
     * 荆州油烟租户id env-test ,tenant_id:41220961c5ad599fd86752a71cff6471,projectId = 20002871
     * 荆州油烟项目id env-prd ,projectId = 20004835,tenant_id: e4f5ab9584f47cbabb4a12b5a014dec9
     */
    private final static List<String> TENANT_IDS = Arrays.asList("e4f5ab9584f47cbabb4a12b5a014dec9");

    private static final Logger LOGGER = LoggerFactory.getLogger(TimingWarningImpl.class);

    private ESconfig eSconfig;
    ThreadPoolExecutor timingWarningThreadPool = new ThreadPoolExecutor(5, 15, 1L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(500), new NamedThreadFactory("timingWarning-ThreadPool"), new ThreadPoolExecutor.AbortPolicy());


    @Resource
    private RedisUtil redisUtil;

    @Autowired
    public void setESconfig(ESconfig eSconfig) {
        this.eSconfig = eSconfig;
    }

    @Resource
    private SmokeWarningService smokeWarningService;
    @Resource
    private LinkappTenantMapper linkappTenantMapper;
    @Resource
    private DeviceService deviceService;
    @Resource
    private DataPushService dataPushService;
    @Resource
    private DeviceUnitService deviceUnitService;

    private List<String> getTenantId() {
        try {
            if (redisUtil.hasKey("jingZhouSmoke_tenant_id")) {
                List list = redisUtil.lGet("jingZhouSmoke_tenant_id", 0, redisUtil.lGetListSize("jingZhouSmoke_tenant_id"));
                return list;
            }
        } catch (Exception e) {
            LOGGER.error("redis 取值错误！", e);
        }
        return TENANT_IDS;
    }

    @Override
    public void execute(JobExecutionContext jobExecutionContext) {
        LOGGER.info("start execute jzyy task ................ start");
        JobDetail jobDetail = jobExecutionContext.getJobDetail();
        // 取得job名称
        String jobName = jobDetail.getKey().getName();
        String jobGroup = jobDetail.getKey().getGroup();
        LOGGER.info("荆州油烟定时任务开始执行了--" + jobGroup + "@" + jobName + DateUtil.getYYYYMMDDHHMMSSDate(new Date()));
        for (String tenantId : getTenantId()) {
            saveCollection(tenantId);
        }
        LOGGER.info("start execute jzyy task ................ end");
    }

    //    @Test
    public void test() {
        for (String tenantId : getTenantId()) {
            saveCollection(tenantId);
        }
    }

    /**
     * 分组查询各设备在周期内的油烟平均值
     */
    private void saveCollection(String tenantId) {
        List<SmokeWarning> smokeWarningList = getCollection(tenantId);
        if (ObjectUtils.isEmpty(smokeWarningList)) {
            LOGGER.info("统计结果为空");
            return;
        }
        Set<String> deviceCodes = smokeWarningList.stream().map(SmokeWarning::getDeviceCode).collect(Collectors.toSet());

        QueryWrapper<Device> qw0 = new QueryWrapper<>();
        qw0.in("code", deviceCodes);
        List<Device> devices = deviceService.list(qw0);
        Map<String, Device> deviceMap = new HashMap<>(600);
        for (Device device : devices) {
            deviceMap.put(device.getCode(), device);
        }
        Set<String> deviceUnitIds = devices.stream().map(Device::getDeviceUnitId).collect(Collectors.toSet());
        QueryWrapper<DeviceUnit> qw1 = new QueryWrapper<>();
        qw1.in("id", deviceUnitIds);
        List<DeviceUnit> deviceUnitList = deviceUnitService.list(qw1);
        Map<String, DeviceUnit> deviceUnitMap = new HashMap<>();
        for (DeviceUnit deviceUnit : deviceUnitList) {
            deviceUnitMap.put(deviceUnit.getId(), deviceUnit);
        }

        List<SmokeWarning> smokeWarningList2 = new ArrayList<>();
        for (SmokeWarning smokeWarning : smokeWarningList) {
            smokeWarning.setCreateTime(new Date());
            Device device = deviceMap.get(smokeWarning.getDeviceCode());
            smokeWarning.setDeviceName(device.getName());
            smokeWarning.setAreaPath(device.getAreaPath());
            try {
                smokeWarningService.save(smokeWarning);
                smokeWarningList2.add(smokeWarning);
            } catch (Exception e) {
                LOGGER.warn("保存荆州油烟平均统计报错：" + e.getMessage());
            }
        }
        datapushSmoke(smokeWarningList2, deviceMap, deviceUnitMap);
    }

    /**
     * 构建流水推送给自己，走规则引擎产生告警
     *
     * @param smokeWarningList2 油烟预警统计列表
     */
    private void datapushSmoke(List<SmokeWarning> smokeWarningList2, Map<String, Device> deviceMap, Map<String, DeviceUnit> deviceUnitMap) {
        if (ObjectUtils.isEmpty(smokeWarningList2)) {
            LOGGER.info("datapushSmoke 为空，不需要推送");
            return;
        }
        for (SmokeWarning smokeWarning : smokeWarningList2) {
            // 监控打印线程状态
            printThreadInfo();
            timingWarningThreadPool.execute(() -> {
                Device device = deviceMap.get(smokeWarning.getDeviceCode());
                DeviceUnit deviceUnit = deviceUnitMap.get(device.getDeviceUnitId());
                DatapushDTO datapushDTO = new DatapushDTO();
                datapushDTO.setDevice_id(device.getCode());
                datapushDTO.setDevice_unit(deviceUnit.getCode());
                datapushDTO.setVersion(deviceUnit.getVersion());
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("average_value", smokeWarning.getAverageValue());
                datapushDTO.setData(jsonObject);
                LOGGER.info("荆州油烟定时采集，推送假流水判断预警报文：{}", JSONObject.toJSONString(datapushDTO));
//            important 计算平均值的时候不更新最后推送时间
//            device.setLastPushTime(device.getLastPushTime());
                dataPushService.updateDeviceAttrStatusAndGenerateAlarm(device, deviceUnit, datapushDTO.getData(), datapushDTO.getTimestamp());
            });

        }
    }

    private void printThreadInfo() {
        int queueSize = timingWarningThreadPool.getQueue().size();
        int activeCount = timingWarningThreadPool.getActiveCount();
        long completedTaskCount = timingWarningThreadPool.getCompletedTaskCount();
        long taskCount = timingWarningThreadPool.getTaskCount();
        LOGGER.info("timingWaring-当前活动线程数：" + activeCount + ",当前排队线程数：" + queueSize + ",执行完成线程数：" + completedTaskCount + ",总线程数：" + taskCount);
    }

    private List<SmokeWarning> getCollection(String tenantId) {
        int index = 0;
        int size = 0;
//         结束时间 开始时间要自己根据当前时间来构建
//        Date now = DateUtil.getDateFromYYYYMMDDHHmmss("2021-12-01 19:00:00");
        Date now = new Date();
        Calendar cal = Calendar.getInstance();
        cal.setTime(now);
//        如果为方便测试的话可以将下两行注释，把定时任务频率调高
        cal.set(Calendar.MINUTE, 0);
        cal.set(Calendar.SECOND, 0);
        cal.set(Calendar.MILLISECOND, 0);
        Date endTime = cal.getTime();
        String endTimeStr = DateUtil.getYYYYMMDDHHMMSSDate(endTime);
        cal.add(Calendar.HOUR_OF_DAY, -1);
        Date startTime = cal.getTime();
        String beginTimeStr = DateUtil.getYYYYMMDDHHMMSSDate(startTime);
        QueryBuilder boolquery = QueryBuilders.boolQuery();
        index--;
        // es 6.0以上支持RestHighLevelClient
        RestHighLevelClient hclient = eSconfig.getFactory().getRhlClient();
//        获取索引,这个月和上个月的 deviceflow-202101,deviceflow-202012
        String indices = EsCalendar.getDeviceFlowIndex(2);
        SearchRequest searchRequest = new SearchRequest(indices);
        //忽略不可用索引，允许索引不不存在，通配符表达式将扩展为打开的索引
        searchRequest.indicesOptions(IndicesOptions.fromOptions(true, true, true, false));
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 查询条件，可以参考官网手册
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        QueryWrapper<LinkappTenant> qw = new QueryWrapper<>();
        qw.select("project_id");
        qw.eq("id", tenantId);
        LinkappTenant tenant = linkappTenantMapper.selectOne(qw);
        Assert.notNull(tenant, "查询到租户为空！");
        boolQuery.must(QueryBuilders.matchPhraseQuery("projectId", tenant.getProjectId()));
        boolQuery.must(boolquery);
        boolQuery.filter(QueryBuilders.rangeQuery("createTime").gte(beginTimeStr).lte(endTimeStr).format("yyyy-MM-dd HH:mm:ss"));
        searchSourceBuilder.query(boolQuery);
        String[] includeFields = new String[]{};
        String[] excludeFields = new String[]{"sourceRef"};
        searchSourceBuilder.fetchSource(includeFields, excludeFields);
        // 取数大小
        searchSourceBuilder.from(index * size).size(size);
        TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("device_code_agg").field("deviceCode").size(30000);
        AvgAggregationBuilder avg = AggregationBuilders.avg("avg_smoke_density")
                .field("data.smoke_density");
        termsAggregationBuilder.subAggregation(avg);
        searchSourceBuilder.aggregation(termsAggregationBuilder);
        searchRequest.source(searchSourceBuilder);
        List<SmokeWarning> result = new ArrayList();
        try {
            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            // 查询结果
            LOGGER.info("设备数据详情ES请求数据：" + searchRequest);
            SearchResponse searchResponse = hclient.search(searchRequest, RequestOptions.DEFAULT);
            stopWatch.stop();
            LOGGER.info("查询用时ms:" + stopWatch.getTotalTimeMillis());
            Aggregations aggregations = searchResponse.getAggregations();
            Terms deviceCodeAgg = aggregations.get("device_code_agg");
            List<? extends Terms.Bucket> bucketAvg = deviceCodeAgg.getBuckets();
            for (Terms.Bucket bucket : bucketAvg) {
                ParsedAvg parsedAvg = bucket.getAggregations().get("avg_smoke_density");
                SmokeWarning sw = new SmokeWarning();
                sw.setDeviceCode(bucket.getKey().toString());
                BigDecimal bigDecimal = BigDecimal.valueOf(parsedAvg.getValue());
                bigDecimal = bigDecimal.setScale(2, BigDecimal.ROUND_HALF_UP);
                sw.setAverageValue(bigDecimal.doubleValue());
                sw.setDataCount(Long.valueOf(bucket.getDocCount()).intValue());
                sw.setTenantId(tenantId);
                sw.setEndTime(endTime);
                sw.setStartTime(startTime);
                result.add(sw);
            }
        } catch (Exception e) {
            LOGGER.error("getDeviceDataListFromES 搜索ES数据,索引:" + indices + "异常:", e);
        }
        return result;
    }

    @Test
    public void test1() {
        BigDecimal bigDecimal = BigDecimal.valueOf(3.4454444);
        bigDecimal = bigDecimal.setScale(2, BigDecimal.ROUND_HALF_UP);
        System.out.println(bigDecimal.doubleValue());
    }

    @Test
    public void test2() {
        Runnable p = () -> {
            for (; ; ) {
                Date now = new Date();
                Calendar cal = Calendar.getInstance();
                cal.setTime(now);
//        如果为方便测试的话可以将下两行注释，把定时任务频率调高
                cal.set(Calendar.MINUTE, 0);
                cal.set(Calendar.SECOND, 0);
                Date endTime = cal.getTime();
                String endTimeStr = DateUtil.getYYYYMMDDHHMMSSDate(endTime);
//            System.out.println(endTimeStr);
                if (!endTimeStr.endsWith("00")) {
                    System.out.println(endTimeStr + "111111");
                    break;
                }
                System.out.println(Thread.currentThread() + endTimeStr);
            }
        };
        for (int i = 0; i < 100; i++) {
            Thread t1 = new Thread(p);
            t1.start();
        }
    }


}

